package org.example.mq.mqserver.core;

import org.example.mq.common.exception.MqException;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.*;

public class ConsumerManager {
    private VirtualHost virtualHost;

    ExecutorService executor = Executors.newFixedThreadPool(4);
    // 队列名通知队列（中心任务调度）
    BlockingQueue<String> notifyQueue = new LinkedBlockingQueue<>();
    // 防止重复通知（可选）
    Set<String> activeQueueSet = ConcurrentHashMap.newKeySet();
    //扫描线程
    Thread scannerThread = null;

    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        scannerThread = new Thread(()->{
            while(true){
                try {
                    String queueName = notifyQueue.take();
                    MessageQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if(queue==null){
                        throw new MqException("[ConsumerManger]队列:"+queueName+"不存在！");
                    }

                    synchronized (queue){
                        //消费一条消息
                        consumeMessage(queue);
                    }

                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    //注册消费者
    public void register(String consumerTag,String channelId,String queueName,boolean autoAck,Consumer callback) throws InterruptedException {
        MessageQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if(queue==null){
            throw new MqException("[ConsumerManager:register]队列:"+queueName+"不存在");
        }

        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,callback,channelId);

        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);

            //将未消费的消息手动消费，不然会消息积压
            int messageCount = virtualHost.getMemoryDataCenter().getMessageCount(queueName);
            if(messageCount!=0){
                for (int i = 0; i < messageCount; i++) {
                    consumeMessage(queue);
                }
            }
        }
    }

    //消费消息
    private void consumeMessage(MessageQueue queue) throws InterruptedException {
        //按照轮询查找消费者
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        //如果无消费者则 return
        if (consumerEnv == null) {
            return;
        }
        //从队列取出消息
        Message message = virtualHost.getMemoryDataCenter().takeMessageFromMessageQueue(queue.getName());
        //将消息加入待确认队列
        virtualHost.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);

        //把消息带入到回调函数中，丢给线程池执行
        executor.submit(() -> {
            try {
                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(),
                        message.getBasicProperties(), message.getBody());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

            //分两种处理消息确认
                    if (consumerEnv.isAutoAck()) {
                        //自动应答,无论消费消息是否有问题直接删除掉
                        if (message.isDelivery()) {
                            virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                        }

                        virtualHost.getMemoryDataCenter().delAckMessage(queue.getName(), message.getMessageId());

                        virtualHost.getMemoryDataCenter().delMessage(message.getMessageId());
                    }

                }
        );
    }

    public void notifyMessage(String queueName){
        if(activeQueueSet.add(queueName)){
            notifyQueue.offer(queueName);
        }
    }
}
